package bma.common.langutil.concurrent;

import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

import bma.common.langutil.core.DateTimeUtil;

/**
 * 基于队列的处理对象
 * 
 * @author 关中
 * @since 1.0 (imported from BMA.ESP Project)
 * 
 */
public class ProcessQueue implements Wakeupable {

	final org.slf4j.Logger log = org.slf4j.LoggerFactory
			.getLogger(ProcessQueue.class);

	public static final long MAX_SLEEP_TIME = 3600 * 1000;
	protected static final Object WAKEUP = new Object();

	protected String name;
	protected BlockingQueue<Object> intentQueue;
	protected volatile boolean closed = true;
	protected volatile boolean processing = false;
	protected transient List<Future4Task<Boolean>> closeFutureList = new LinkedList<Future4Task<Boolean>>();

	protected boolean skipLogWait = true;

	public ProcessQueue() {
		this(null, null);
	}

	public ProcessQueue(String name) {
		this(name, null);
		this.name = name;
	}

	public ProcessQueue(String name, BlockingQueue<Object> intentQueue) {
		super();
		this.name = name == null ? "PQ" + System.identityHashCode(this) : name;
		this.intentQueue = intentQueue == null ? new LinkedBlockingQueue<Object>()
				: intentQueue;
	}

	public String getName() {
		return name;
	}

	public void setName(String name) {
		this.name = name;
	}

	public int queueSize() {
		return intentQueue.size();
	}

	public int load() {
		return intentQueue.size() + (processing ? 1 : 0);
	}

	/**
	 * 处理内部事件
	 * 
	 * @param event
	 */
	protected void processInternalEvent(Object event) {
		if (event instanceof Runnable) {
			Runnable r = (Runnable) event;
			r.run();
		}
	}

	public boolean initProcess() {
		closed = false;
		return true;
	}

	public boolean isAlive() {
		if (closed) {
			if (intentQueue.isEmpty()) {
				return false;
			}
		}
		return true;
	}

	public void shutdown() {
		if (!closed) {
			throw new IllegalStateException("close() before shutdown");
		}
		if (log.isDebugEnabled()) {
			log.debug("{}: cleanup", name);
		}
		this.intentQueue.clear();

		for (Future4Task<Boolean> f : this.closeFutureList) {
			f.done(true, null);
		}
	}

	public void process(long waitTime) {
		doprocess(waitTime);
	}

	public boolean postIntent(Object intent) {
		if (intent == null)
			return false;
		if (closed) {
			return false;
		}
		this.intentQueue.add(intent);
		return true;
	}

	protected void doprocess(long waitTime) {
		long sleep = 0;
		long waitTo = waitTime == 0 ? 0 : waitTime + System.currentTimeMillis();
		while (true) {
			if (closed) {
				// closing,don't sleep
				sleep = 0;
			} else {
				// check for waitTime
				if (waitTo > 0) {
					long maxSleep = waitTo - System.currentTimeMillis();
					if (maxSleep <= 0)
						return;
					// if (sleep == 0) {
					// sleep = maxSleep;
					// } else
					if (sleep > maxSleep) {
						sleep = maxSleep;
					}
				}
			}
			Object intent = null;
			if (sleep > 0) {
				if (log.isDebugEnabled() && !skipLogWait) {
					log.debug(
							"{}: wait {}",
							new Object[] { name,
									DateTimeUtil.formatPeriod(sleep) });
				}
				try {
					intent = this.intentQueue
							.poll(sleep, TimeUnit.MILLISECONDS);
				} catch (InterruptedException e) {
					// do nothing
				}
			} else {
				intent = this.intentQueue.poll();
			}

			if (intent == null) {
				if (closed) {
					// empty & closed
					return;
				}
			} else {
				processing = true;
				try {
					if (intent == WAKEUP) {
						return;
					}
					if (intent instanceof ProcessIntentClose) {
						if (log.isDebugEnabled()) {
							log.debug("{}: receive CloseIntent", name);
						}
						try {
							this.close();
						} finally {
							ProcessIntentClose closeIntent = (ProcessIntentClose) intent;
							Future4Task<Boolean> f = closeIntent.getFuture();
							if (f != null) {
								this.closeFutureList.add(f);
							}
						}
					} else {
						if (!this.handleIntent(intent)) {
							return;
						}
					}
				} finally {
					processing = false;
				}
			}

			long sleepTo = processRun();

			if (sleepTo < 0) {
				sleep = MAX_SLEEP_TIME;
			} else if (sleepTo == 0) {
				sleep = 0;
			} else {
				sleep = sleepTo - System.currentTimeMillis();
				if (sleep < 0)
					sleep = 0;
				if (sleep > MAX_SLEEP_TIME)
					sleep = MAX_SLEEP_TIME;
			}
		}
	}

	/**
	 * 处理事件
	 * 
	 * @param intent
	 * @return
	 */
	public boolean handleIntent(Object intent) {
		if (intent == null)
			return true;
		processInternalEvent(intent);
		return true;
	}

	public long processRun() {
		return -1;
	}

	/**
	 * 内部尝试关闭
	 * 
	 * @return
	 */
	protected void close() {
		if (closed) {
			return;
		}
		closed = true;
		if (log.isInfoEnabled()) {
			log.info("{}: closed", name);
		}
	}

	@Override
	public void wakup() {
		postIntent(WAKEUP);
	}

	@Override
	public String toString() {
		return "ProcessQueue(" + name + ")";
	}
}
